/*
 * Package : mqtt_client
 * Author : S. Hamblett <steve.hamblett@linux.com>
 * Date   : 31/05/2017
 * Copyright :  S.Hamblett
 */

import 'dart:async';
import 'dart:convert';
import 'dart:io';

import 'package:flutter/cupertino.dart';
import 'package:mqtt_client/mqtt_client.dart';
import 'package:mqtt_client/mqtt_server_client.dart';
import 'package:wch/constant/event_bus_global.dart';
import 'package:wch/model/deviceInfo.dart';
import 'package:wch/model/device_upgrade_model.dart';
import 'package:wch/model/recycle_device_model.dart';
import 'package:wch/networking/api.dart';
import 'package:wch/utils/log_utils.dart';

/// An annotated simple subscribe/publish usage example for mqtt_server_client. Please read in with reference
/// to the MQTT specification. The example is runnable, also refer to test/mqtt_client_broker_test...dart
/// files for separate subscribe/publish tests.

/// First create a client, the client is constructed with a broker name, client identifier
/// and port if needed. The client identifier (short ClientId) is an identifier of each MQTT
/// client connecting to a MQTT broker. As the word identifier already suggests, it should be unique per broker.
/// The broker uses it for identifying the client and the current state of the client. If you don’t need a state
/// to be hold by the broker, in MQTT 3.1.1 you can set an empty ClientId, which results in a connection without any state.
/// A condition is that clean session connect flag is true, otherwise the connection will be rejected.
/// The client identifier can be a maximum length of 23 characters. If a port is not specified the standard port
/// of 1883 is used.
/// If you want to use websockets rather than TCP see below.

final client = MqttServerClient(debugMqttBaseUrl, '');

var pongCount = 0; // Pong counter

Future<int> connectMqtt() async {
  /// A websocket URL must start with ws:// or wss:// or Dart will throw an exception, consult your websocket MQTT broker
  /// for details.
  /// To use websockets add the following lines -:
  /// client.useWebSocket = true;
  /// client.port = 80;  ( or whatever your WS port is)
  /// There is also an alternate websocket implementation for specialist use, see useAlternateWebSocketImplementation
  /// Note do not set the secure flag if you are using wss, the secure flags is for TCP sockets only.
  /// You can also supply your own websocket protocol list or disable this feature using the websocketProtocols
  /// setter, read the API docs for further details here, the vast majority of brokers will support the client default
  /// list so in most cases you can ignore this.

  /// Set logging on if needed, defaults to off
  client.logging(on: true);

  /// Set the correct MQTT protocol for mosquito
  client.setProtocolV311();

  /// If you intend to use a keep alive you must set it here otherwise keep alive will be disabled.
  client.keepAlivePeriod = 200;

  /// The connection timeout period can be set if needed, the default is 5 seconds.
  client.connectTimeoutPeriod = 10000; // milliseconds

  /// Add the unsolicited disconnection callback
  client.onDisconnected = onDisconnected;

  /// Add the successful connection callback
  client.onConnected = onConnected;

  /// Add a subscribed callback, there is also an unsubscribed callback if you need it.
  /// You can add these before connection or change them dynamically after connection if
  /// you wish. There is also an onSubscribeFail callback for failed subscriptions, these
  /// can fail either because you have tried to subscribe to an invalid topic or the broker
  /// rejects the subscribe request.
  client.onSubscribed = onSubscribed;

  /// Set a ping received callback if needed, called whenever a ping response(pong) is received
  /// from the broker.
  client.pongCallback = pong;

  /// Create a connection message to use or use the default one. The default one sets the
  /// client identifier, any supplied username/password and clean session,
  /// an example of a specific one below.
  final connMess = MqttConnectMessage()
      .withClientIdentifier('Mqtt_MyClientUniqueId')
      .withWillTopic(
      'RL/SAAS/FLEX_AIO_22330C') // If you set this you must set a will message
      .withWillMessage('flutter hello 3')
      .startClean() // Non persistent session for testing
      .withWillQos(MqttQos.exactlyOnce);
  print('EXAMPLE::Mosquitto client connecting....');
  // client.connectionMessage = connMess;

  /// Connect the client, any errors here are communicated by raising of the appropriate exception. Note
  /// in some circumstances the broker will just disconnect us, see the spec about this, we however will
  /// never send malformed messages.
  try {
    await client.connect();
  } on NoConnectionException catch (e) {
    // Raised by the client when connection fails.
    print('EXAMPLE::client exception - $e');
    // client.disconnect();
  } on SocketException catch (e) {
    // Raised by the socket layer
    print('EXAMPLE::socket exception - $e');
    // client.disconnect();
  }

  /// Check we are connected
  if (client.connectionStatus!.state == MqttConnectionState.connected) {
    print('EXAMPLE::Mosquitto client connected');
  } else {
    /// Use status here rather than state if you also want the broker return code.
    print(
        'EXAMPLE::ERROR Mosquitto client connection failed - disconnecting, status is ${client
            .connectionStatus}');
    client.disconnect();
    // exit(-1);
  }

  /// Ok, lets try a subscription
  print('EXAMPLE::Subscribing to the test/lol topic');
  const topic = 'test/lol'; // Not a wildcard topic
  client.subscribe(topic, MqttQos.atMostOnce);

  /// The client has a change notifier object(see the Observable class) which we then listen to to get
  /// notifications of published updates to each subscribed topic.
  /// In general you should listen here as soon as possible after connecting, you will not receive any
  /// publish messages until you do this.
  /// Also you must re-listen after disconnecting.
  client.updates!.listen((List<MqttReceivedMessage<MqttMessage?>>? c) {
    final recMess = c![0].payload as MqttPublishMessage;
    // final pt =
    // MqttPublishPayload.bytesToStringAsString(recMess.payload.message);

    // 将 GBK 编码的字符串转换回 Uint8List
    // Uint8List.fromList(recMess.payload.message);
    // Uint8List bytes = Uint8List.fromList(recMess.payload.message); // 这是 "你好" 的 UTF-8 编码

    // 将 Uint8List 转换为 GBK 编码的字符串
    final pt = utf8.decode(recMess.payload.message);

    /// The above may seem a little convoluted for users only interested in the
    /// payload, some users however may be interested in the received publish message,
    /// lets not constrain ourselves yet until the package has been in the wild
    /// for a while.
    /// The payload is a byte buffer, this will be specific to the topic
    print(
        'EXAMPLE::Change notification:: topic is <${c[0]
            .topic}>, payload is <-- $pt -->');
    print('');
  });

  /// If needed you can listen for published messages that have completed the publishing
  /// handshake which is Qos dependant. Any message received on this stream has completed its
  /// publishing handshake with the broker.
  setPublishListener();

  /// Lets publish to our topic
  /// Use the payload builder rather than a raw buffer
  /// Our known topic to publish to
  // const pubTopic = 'RL/SAAS/FLEX_AIO_22330C';
  // final builder = MqttClientPayloadBuilder();
  // builder.addString('Hello from mqtt_client2');
  //
  // /// Subscribe to it
  // print('EXAMPLE::Subscribing to the RL/SAAS/FLEX_AIO_22330C');
  // client.subscribe(pubTopic, MqttQos.exactlyOnce);
  //
  // /// Publish it
  // print('EXAMPLE::Publishing our topic');
  // client.publishMessage(pubTopic, MqttQos.exactlyOnce, builder.payload!);
  //
  // /// Ok, we will now sleep a while, in this gap you will see ping request/response
  // /// messages being exchanged by the keep alive mechanism.
  // print('EXAMPLE::Sleeping....');
  // // await MqttUtilities.asyncSleep(60);
  //
  // /// Finally, unsubscribe and exit gracefully
  // print('EXAMPLE::Unsubscribing');
  // // client.unsubscribe(topic);
  //
  // /// Wait for the unsubscribe message from the broker if you wish.
  // // await MqttUtilities.asyncSleep(2);
  // print('EXAMPLE::Disconnecting');
  // // client.disconnect();
  // print('EXAMPLE::Exiting normally');
  return 0;
}

void setPublishListener() {
  client.published!.listen((MqttPublishMessage message) {
    print(
        'EXAMPLE::Published notification:: topic is ${message.variableHeader!
            .topicName}, with Qos ${message.header!.qos}');
    if(message.variableHeader!.topicName.contains('/APP')) {
      return;
    }

    ///收到订阅消息
    final pt =
    MqttPublishPayload.bytesToStringAsString(message.payload.message);
    Map<String, dynamic> covertMsg = jsonDecode(pt);
    debugPrint('covertMsg=$covertMsg');

    if (covertMsg['data'].runtimeType == List) {
      List<dynamic> list = covertMsg['data'];
      debugPrint('list.len=${list.length}');
      if (list.isNotEmpty) {
        Map<String, dynamic> device1 = list.first;

        String aioName = device1['deviceName'];
        String aioType = device1['deviceType'];
        String aioVersion = device1['deviceVersion'];
        String aioWorkStatus = device1['workStatus'];
        String aioUpgrade = device1['upgradeStatus'];
        String aioWifiName = device1['wifiName'];
        String aioWifiStatus = device1['wifiStatus'];

        String? deviceName;
        String? deviceType;
        String? deviceVersion;
        String? signal;
        String? battery;
        String? workStatus;
        String? upgradeStatus;
        String? connStatus;

        if (list.length >= 2) {
          Map<String, dynamic> device2 = list.last;
          deviceName = device2['deviceName'];
          deviceType = device2['deviceType'];
          deviceVersion = device2['deviceVersion'];
          signal = device2['signal'];
          battery = device2['battery'];
          workStatus = device2['workStatus'];
          upgradeStatus = device2['upgradeStatus'];
          connStatus = device2['connStatus'];
        }

        DeviceInfoModel model = DeviceInfoModel(
            aioName,
            aioType,
            aioVersion,
            aioWorkStatus,
            aioUpgrade,
            aioWifiName,
            aioWifiStatus,
            deviceName,
            deviceType,
            deviceVersion,
            signal,
            battery,
            workStatus,
            upgradeStatus,
            connStatus
        );
        globalEventBus.fire(model);
      }
    } else if (covertMsg['messageType'] == 'collect_suc') {///可以回收设备
      globalEventBus.fire(RecycleDeviceModel(result: true));
    } else if (covertMsg['messageType'] == 'collect_fail') {///不可回收设备
      globalEventBus.fire(RecycleDeviceModel(result: false));
    } else if (covertMsg['messageType'] == 'aio_upgrade') {///一体机升级状态
      Map<String, dynamic> data = covertMsg['data'];
      // globalEventBus.fire(DeviceUpgradeModel(status: int.parse(data['upgradeStatus']), type: 1));
      globalEventBus.fire(DeviceUpgradeModel(status: data['upgradeStatus'], type: 1));
    } else if (covertMsg['messageType'] == 'eeg_upgrade') {///额贴升级状态
      Map<String, dynamic> data = covertMsg['data'];
      // globalEventBus.fire(DeviceUpgradeModel(status: int.parse(data['upgradeStatus']), type: 2));
      globalEventBus.fire(DeviceUpgradeModel(status: data['upgradeStatus'], type: 2));
    }
  });
}

/// The subscribed callback
void onSubscribed(String topic) {
  print('EXAMPLE::Subscription confirmed for topic $topic');
}

/// The unsolicited disconnect callback
void onDisconnected() {
  print('EXAMPLE::OnDisconnected client callback - Client disconnection');
  if (client.connectionStatus!.disconnectionOrigin ==
      MqttDisconnectionOrigin.solicited) {
    print('EXAMPLE::OnDisconnected callback is solicited, this is correct');
    reConnect();
  } else {
    print(
        'EXAMPLE::OnDisconnected callback is unsolicited or none, this is incorrect - exiting');
    // exit(-1);
    reConnect();
  }
}

void reConnect() async{
  debugPrint('reConnect');
  try {
    await client.connect();

    setPublishListener();
  } on NoConnectionException catch (e) {
    // Raised by the client when connection fails.
    print('EXAMPLE::client exception - $e');
    // client.disconnect();
    ///5秒钟后做重连尝试
    LogUtils.log('-重连 mqtt-');
    Timer(Duration(seconds: 5), (){
      reConnect();
    });
  } on SocketException catch (e) {
    // Raised by the socket layer
    print('EXAMPLE::socket exception - $e');
    // client.disconnect();
    ///5秒钟后做重连尝试
    LogUtils.log('-重连 mqtt-');
    Timer(Duration(seconds: 5), (){
      reConnect();
    });
  }
}

/// The successful connect callback
void onConnected() {
  print(
      'EXAMPLE::OnConnected client callback - Client connection was successful');
}

/// Pong callback
void pong() {
  print('EXAMPLE::Ping response client callback invoked');
  pongCount++;
}

///关闭 mqtt 连接
void disConnectMqtt() {
  if (client != null) {
    client.disconnect();
  }
}

///订阅主题
void mqttSubscribe(String topic) {
  // const pubTopic = 'RL/SAAS/FLEX_AIO_22330C';
  /// Subscribe to it
  print('EXAMPLE::Subscribing to the $topic');
  client.subscribe(topic, MqttQos.exactlyOnce);

  /// Publish it
  print('EXAMPLE::Publishing our topic');
  // client.publishMessage(pubTopic, MqttQos.exactlyOnce, builder.payload!);
  /// Wait for the unsubscribe message from the broker if you wish.
}

void mqttCancelSubscribe(String topic) {
  client.unsubscribe(topic);
  debugPrint('解除订阅$topic');
}

///发布消息
void mqttPublicMsg(String pubTopic, String msg) {
  final builder = MqttClientPayloadBuilder();
  builder.addString(msg);
  /// Publish it
  debugPrint('mqttPublicMsg = $msg');
  client.publishMessage(pubTopic, MqttQos.exactlyOnce, builder.payload!);
}
